package com.gdlt.mq2db;

import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;
import java.lang.reflect.Constructor;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.hadoop.conf.Configuration;
import org.apache.kafka.clients.consumer.ConsumerRecord;

/**
 * Created by CM on 2017/3/14.
 *
 */

public class Runner {

	static long offset = 0;
	static KafkaReader reader = null;
	 static BlockingQueue<ConsumerRecord<String, String>> queue;
	
	public static void main(String[] args) throws Exception {

		int partitionId = Integer.parseInt(args[0]);// read partitionid and config from args.
		String configFile = args[1];
		InputStream inputStream = new FileInputStream(new File(configFile));

		Kafka2pgConfig.AddConfig(inputStream);
		Kafka2pgConfig.AddArgs(partitionId);

		Configuration conf = Kafka2pgConfig.getConf(); // get the value from the xml file.

		String writeClasses[] = conf.getStrings(Kafka2pgConfig.WRITER_CLASSES_CONF_NAME);
		int blockingQueueSize = conf.getInt(Kafka2pgConfig.BLOCKINGQUEUE_SIZE_CONF_NAME, Kafka2pgConfig.BLOCKINGQUEUE_SIZE_DEFAULT_VALUE);
		int classLen = writeClasses.length;
		final Writable[] writables = new Writable[classLen];
		
		queue= new LinkedBlockingQueue<ConsumerRecord<String,String>>(blockingQueueSize);
		
		
		for (int i = 0; i < classLen; i++) {
			Class<?> writers = Class.forName(writeClasses[i]);
			 Constructor<?> constructor = writers.getConstructor(BlockingQueue.class);//有参构造函数
			 Object object =  constructor.newInstance(queue);
			Writable writable = (Writable) object;
			System.out.println(writable);
			writables[i] = writable;

			if (writable instanceof Offsetable) {
				Offsetable offsetable = (Offsetable) object;
				offset = offsetable.getLastoffset();
				System.out.println(offset);
			}
		}
		
		reader = new KafkaReader(partitionId, offset,queue);
		

		Runtime.getRuntime().addShutdownHook(new Thread() {
			@Override
			public void run() {
				System.out.println("Shutdown hook ran!");
				reader.close();
				for (int i = 0; i < writables.length; i++) {
					writables[i].close();
				}
			}
		});


		//while (reader.isClosed()) {
			if (reader.isClosed()) {
			   ExecutorService service = Executors.newCachedThreadPool();//线程池
			   service.submit(reader);//启动消费线程			
			//BlockingQueue<ConsumerRecord<String, String>> data = null;
			  // List<ConsumerRecord<String, String>> data = reader.getData();
			  // offset = reader.getOffset();
          
			for (int i = 0; i < writables.length; i++) {
				if (writables[i].isClosed())		
					service.submit((Thread) writables[i]);		 
					//writables[i].write(data, offset);
				
			}
		}

	}

}
